package com.lei.mq;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 延迟消息实现,、
 * 延迟消息必须通过两个队列实现，通过队列1消息过期后发送给队列2实现延迟消息
 * 官网：http://www.rabbitmq.com/dlx.html
 * Created by Administrator on 2019/2/26.
 */
@Component
public class S06ConfirmQueueProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    //路由key
    private static final String RUOTE_KEY = "s06.key";
    private static final String EXCHANGE = "s06.exchange";
    private static final String QUEUE = "s06expiration.queue";

    /**
     * 定义一个exchange交换机
     *
     * @return
     */
    @Bean
    public DirectExchange s06ExpirationStatics() {
        return new DirectExchange(EXCHANGE);
    }

    /**
     * 定义一个队列
     *
     * @return
     */
    @Bean
    public Queue s06ExpirationQueue() {
        return new Queue(QUEUE);
    }

    @Bean
    public Binding s06BindingExpiration() {
        return BindingBuilder.bind(s06ExpirationQueue()).to(s06ExpirationStatics()).with(RUOTE_KEY);
    }

    @Scheduled(fixedDelay = 2000L)
    public void send() {
        System.out.println("S06ConfirmQueueProducer");
        SimpleDateFormat format = new SimpleDateFormat("yyyy-mm-dd hh:mm:ss");
        String content = format.format(new Date());
        //需要配置 publisher-confirms: true，回调方法才生效
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            System.out.println("confirm--:correlationData:" + correlationData + ",ack:" + ack + ",cause:" + cause);
            /**
             * 如果消息没有到exchange,则confirm回调,ack=false
             * 如果消息到达exchange,则confirm回调,ack=true
             */
            if (ack) {
                System.out.println("发送成功");
            } else {
                //发送过程中删除交换机，或服务器没有断开了会发送失败，但不清楚是那个交换机失败
                System.out.println("发送失败");
            }
        });
        //需要配置 publisher-returns: true，回调方法才生效
        rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText,
                                          String exchange, String routingKey) ->{
                /**
                 * exchange到queue成功,则不回调return
                 * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
                 */
                System.out.println("returnCallback--:message:" + message + ",replyCode:" + replyCode + ",exchange:" + exchange + ",routingKey:" + routingKey);
                System.out.println(message.getMessageProperties().getHeaders().get("spring_returned_message_correlation"));
                }
        );
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(UUID.randomUUID().toString().replace("-", ""));
        rabbitTemplate.convertSendAndReceive(EXCHANGE, RUOTE_KEY, "S06ConfirmQueueProducer " + content, correlationData);
    }
}
